package march.finalmarch22.march02;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class StuRecordWriter extends RecordWriter<Text, Text> {

    private FSDataOutputStream mathOut;
    private FSDataOutputStream chinOut;
    private FSDataOutputStream engOut;

    public StuRecordWriter(TaskAttemptContext job){

        try{
            FileSystem fs = FileSystem.get(job.getConfiguration());
            mathOut = fs.create(new Path("F:\\test00\\Math\\math"));
            chinOut = fs.create(new Path("F:\\test00\\Chinese\\chinese"));
            engOut = fs.create(new Path("F:\\test00\\English\\english"));
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    @Override
    public void write(Text key, Text value) throws IOException, InterruptedException {
        String score = key.toString();
        if(score.contains("Chinese")){
            chinOut.writeBytes(score+"\n");
        }else if (score.contains("Math")){
            mathOut.writeBytes(score+"\n");
        }else{
            engOut.writeBytes(score+"\n");
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {

        IOUtils.closeStream(mathOut);
        IOUtils.closeStream(engOut);
        IOUtils.closeStream(chinOut);
    }
}
